package com.cskaoyan.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.cskaoyan.mall.common.constant.RedisConst;
import com.cskaoyan.mall.mq.constant.MqResultEnum;
import com.cskaoyan.mall.promo.constant.LocalStockStatus;
import com.cskaoyan.mall.promo.mapper.SeckillGoodsMapper;
import com.cskaoyan.mall.promo.redis.RedisStockOper;
import com.cskaoyan.mall.promo.util.LocalCacheHelper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RBucket;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;

/**
 * @Author Pudding
 * @Date 2024/7/23 17:07
 * 事务消息的生成者
 */
@Slf4j
@Component
public class PromoTransactionProducer {

    TransactionMQProducer txProducer;

    @Autowired
    RedissonClient redissonClient;
    @Autowired
    SeckillGoodsMapper seckillGoodsMapper;
    @Autowired
    RedisStockOper redisStockOper;

    //1. 初始化事务消息的生产者
    @PostConstruct
    public void init() throws MQClientException {

        //1. 创建事务消息的Producer
        txProducer = new TransactionMQProducer("tx-promo-service-group");

        //2. 设置注册中心
        txProducer.setNamesrvAddr("47.96.11.99:9876");

        //3. 设置事务监听器
        txProducer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地事务 ——> 其实就是扣减秒杀库存
             * @param msg   事务消息
             * @param arg   业务参数
             * @return      本地事务的执行结果：COMMIT_MESSAGE：本地事务执行成功、ROLLBACK_MESSAGE：本地事务执行失败、UNKNOW：本地事务执行结果未知
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {

                //1. 获取本地事务的执行参数，这个参数是下面方法传过来的localTransactionParam，这个是一个map
                HashMap<String, Object> hashMap = (HashMap<String, Object>) arg;
                Long id = (Long) hashMap.get("id");
                Integer stock = (Integer) hashMap.get("stock");
                Long skuId = (Long) hashMap.get("skuId");

                //获取消息的id，是为了检查本地事务的执行结果：库存扣减成功没有，用消息id：transactionId作为redis中的key打一个标记
                String transactionId = msg.getTransactionId();
                RBucket<String> bucket = redissonClient.getBucket(RedisConst.PROMO_TRANSACTION_PREFIX + transactionId);

                //2. 执行本地事务 ——> 其实就是扣减秒杀库存
                /*
                方案一：
                    int affectedRows = seckillGoodsMapper.decreaseStock(id, stock);
                    //2.1 说明本地事务执行失败，返回本地事务执行失败的结果
                    if (affectedRows < 1){

                        //打一个标记，标记库存扣减成功与否
                        bucket.set("fail");
                        //更新本地库存状态位标记
                        LocalCacheHelper.put(skuId.toString(), LocalStockStatus.NO_STOCK.getNum());

                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                */

                //方案二：
                //改进：之前是扣减数据库中的库存，现在由于库存已经存储到了Redis，所以在这里，可以直接扣减Redis中的库存，减库存效率更高

                //如何扣减 Redis中的库存呢?
                //方式一：先查询出Redis中库存，然后扣减，再把扣减后的库存存入Redis
                //上述代码，存在多线程的安全问题：因为上述操作，是三步操作，不是一个原子操作，意味着在多线程的情况下，存在数据安全问题―→超卖：库存只扣减一次，但是生成了多个订单

                //方式二：在第一个方案的基础之上，进行加锁可以实现，但是效率比较低

                //方式三：使用hincrby命令，来扣减库存，但是单纯使用该命令来扣减库存，可以减到负数，但是库存不能为负数

                //方式四：还是使用hincrby命令，来扣减库存，在调用命令扣减库存之后，先查询库存，判断结果是否为负数。但是扣减库存和判断库存不是一个原子性操作，也就是存在多线程的数据安全问题

                //方式五：使用lua脚本，保证redis中的判断库存和扣减库存为一个原子性操作
                //2.1.1 因为Redis的工作线程是单线程，且Lua脚本可以直接在Redis中运行，所以一段Lua脚本中运行的必然是一个原子操作
                Long remainStock = redisStockOper.decrRedisStock(skuId, stock);
                //2.1.2 当前扣减库存，正好扣减完成
                if (remainStock.intValue() <= 0){
                    //在redis添加消息消费失败的标记
                    bucket.set("fail");
                    //更新本地库存状态位标记
                    LocalCacheHelper.put(skuId.toString(), LocalStockStatus.NO_STOCK.getNum());
                    //返回本地事务执行失败
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                //打一个标记，标记库存扣减成功与否
                bucket.set("success");
                //2.2 说明本地事务执行成功，返回本地事务执行成功的结果
                return LocalTransactionState.COMMIT_MESSAGE;
            }

            /**
             * 检查本地事务的执行结果 ——> 其实就是去看库存扣减没有，在LocalTransactionState.UNKNOW：本地事务执行结果未知情况下执行
             * @param msg      消息对象
             * @return         本地事务的执行结果
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {

                //获取消息id
                String transactionId = msg.getTransactionId();

                RBucket<String> bucket = redissonClient.getBucket(RedisConst.PROMO_TRANSACTION_PREFIX + transactionId);

                //检查扣减秒杀库存时打的标记是否是扣减库存成功
                String value = bucket.get();
                if ("success".equals(value)){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                if ("fail".equals(value)){
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }

                return LocalTransactionState.UNKNOW;
            }
        });

        //4. 启动
        txProducer.start();
        log.info("秒杀事务消息生产者启动成功...");

    }

    /**
     * 2. 提供一个方法，发送事务消息 ———> 之后就会执行本地事务方法
     * @param topic                 消息的主题
     * @param messageObj            消息的内容
     * @param localTransactionParam 执行本地事务，接受的参数，也就是传递给上面执行本地事务方法中的参数arg
     * @return
     */
    public MqResultEnum sendTransactionMessage(String topic, Object messageObj, Object localTransactionParam) {

        try {
            //1. 创建消息对象
            String messageJSON = JSON.toJSONString(messageObj);
            byte[] messageJSONBytes = messageJSON.getBytes(StandardCharsets.UTF_8);
            Message message = new Message(topic, messageJSONBytes);

            //2. 发送事务消息
            TransactionSendResult transactionSendResult = txProducer.sendMessageInTransaction(message, localTransactionParam);

            //3. 获取结果
            SendStatus sendStatus = transactionSendResult.getSendStatus();      //消失的发送结果
            LocalTransactionState localTransactionState = transactionSendResult.getLocalTransactionState(); //本地事务的执行结果

            //4. 对发送的结果进行判断
            //4.1 说明事务消息发送失败
            if (!SendStatus.SEND_OK.equals(sendStatus)){
                return MqResultEnum.SEND_FAIL;
            }
            //4.2 说明消息发送成功，本地事务执行失败
            else if (LocalTransactionState.ROLLBACK_MESSAGE.equals(localTransactionState)){
                return MqResultEnum.LOCAL_TRANSACTION_FAIL;
            }
            //4.3 说明消息发送成功，本地事务执行成功
            else if (LocalTransactionState.COMMIT_MESSAGE.equals(localTransactionState)){
                return MqResultEnum.Local_TRANSACTION_SUCCESS;
            }
            //4.4 如果是其他情况，没有获取到本地事务的执行结果，返回本地事务执行异常
            else {
                return MqResultEnum.LOCAL_TRANSACTION_EXCEPTION;
            }

        } catch (Exception e) {
            e.printStackTrace();
            //如果发送消息发生了异常，返回消息发送失败
            return MqResultEnum.SEND_FAIL;
        }
    }
}
